package workflow

import (
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/asynkron/protoactor-go/actor"
	"github.com/louloulin/dataflare/pkg/actor/system"
)

// WorkflowActor is an actor that executes a workflow instance
type WorkflowActor struct {
	// workflow is the workflow definition
	workflow *WorkflowDefinition
	// instance is the workflow instance
	instance *WorkflowInstance
	// stateManager is the state manager PID
	stateManager *actor.PID
	// metricsCollector is the metrics collector PID
	metricsCollector *actor.PID
	// nodeActors are the node actors
	nodeActors map[string]*actor.PID
	// nodeStatuses are the node statuses
	nodeStatuses map[string]string
	// mu is a mutex for protecting the node actors and statuses
	mu sync.RWMutex
	// logger is the workflow actor logger
	logger *log.Logger
	// supervisor is the workflow supervisor PID
	supervisor *actor.PID
}

// NewWorkflowActor creates a new workflow actor
func NewWorkflowActor(workflow *WorkflowDefinition, instance *WorkflowInstance, stateManager *actor.PID, metricsCollector *actor.PID) actor.Actor {
	return &WorkflowActor{
		workflow:         workflow,
		instance:         instance,
		stateManager:     stateManager,
		metricsCollector: metricsCollector,
		nodeActors:       make(map[string]*actor.PID),
		nodeStatuses:     make(map[string]string),
		logger:           log.New(log.Writer(), fmt.Sprintf("[WorkflowActor-%s] ", instance.ID), log.LstdFlags),
	}
}

// Receive handles messages for the workflow actor
func (w *WorkflowActor) Receive(context actor.Context) {
	switch msg := context.Message().(type) {
	case *actor.Started:
		w.handleStarted(context)
	case *actor.Stopping:
		w.handleStopping(context)
	case *actor.Terminated:
		w.handleTerminated(context, msg)
	case *NodeStatusChangedMessage:
		w.handleNodeStatusChanged(context, msg)
	case *NodeErrorMessage:
		w.handleNodeError(context, msg)
	case *system.DataMessage:
		w.handleDataMessage(context, msg)
	default:
		w.logger.Printf("Received unknown message: %T", msg)
	}
}

// handleStarted handles the Started message
func (w *WorkflowActor) handleStarted(context actor.Context) {
	w.logger.Println("Workflow actor started")

	// Store supervisor
	w.supervisor = context.Parent()

	// Report metrics
	if w.metricsCollector != nil {
		context.Send(w.metricsCollector, &system.ReportMetricCommand{
			Name:  "workflow.actor.started",
			Value: 1,
			Tags: map[string]string{
				"workflow_id": w.workflow.ID,
				"instance_id": w.instance.ID,
			},
		})
	}

	// Create node actors
	err := w.createNodeActors(context)
	if err != nil {
		w.logger.Printf("Failed to create node actors: %v", err)
		w.failWorkflow(context, err.Error())
		return
	}

	// Connect nodes
	err = w.connectNodes(context)
	if err != nil {
		w.logger.Printf("Failed to connect nodes: %v", err)
		w.failWorkflow(context, err.Error())
		return
	}

	// Start source nodes
	err = w.startSourceNodes(context)
	if err != nil {
		w.logger.Printf("Failed to start source nodes: %v", err)
		w.failWorkflow(context, err.Error())
		return
	}

	w.logger.Println("Workflow started successfully")
}

// handleStopping handles the Stopping message
func (w *WorkflowActor) handleStopping(context actor.Context) {
	w.logger.Println("Workflow actor stopping")

	// Stop all node actors
	for name, pid := range w.nodeActors {
		w.logger.Printf("Stopping node actor: %s", name)
		context.Stop(pid)
	}

	// Report metrics
	if w.metricsCollector != nil {
		context.Send(w.metricsCollector, &system.ReportMetricCommand{
			Name:  "workflow.actor.stopped",
			Value: 1,
			Tags: map[string]string{
				"workflow_id": w.workflow.ID,
				"instance_id": w.instance.ID,
			},
		})
	}
}

// handleTerminated handles the Terminated message
func (w *WorkflowActor) handleTerminated(context actor.Context, msg *actor.Terminated) {
	w.mu.Lock()
	defer w.mu.Unlock()

	// Find the terminated node actor
	for name, pid := range w.nodeActors {
		if pid.String() == msg.Who.String() {
			w.logger.Printf("Node actor terminated: %s", name)
			
			// Update node status
			w.nodeStatuses[name] = "terminated"
			
			// Check if all nodes are terminated
			allTerminated := true
			for _, status := range w.nodeStatuses {
				if status != "terminated" && status != "completed" && status != "failed" {
					allTerminated = false
					break
				}
			}
			
			if allTerminated {
				w.logger.Println("All nodes terminated, completing workflow")
				w.completeWorkflow(context)
			}
			
			break
		}
	}
}

// handleNodeStatusChanged handles the NodeStatusChangedMessage message
func (w *WorkflowActor) handleNodeStatusChanged(context actor.Context, msg *NodeStatusChangedMessage) {
	w.mu.Lock()
	defer w.mu.Unlock()

	// Update node status
	w.nodeStatuses[msg.NodeID] = msg.Status

	w.logger.Printf("Node status changed: %s -> %s", msg.NodeID, msg.Status)

	// Check if all nodes are completed
	if msg.Status == "completed" {
		allCompleted := true
		for name, status := range w.nodeStatuses {
			// Skip source nodes, they run continuously
			if _, ok := w.workflow.Sources[name]; ok {
				continue
			}
			
			if status != "completed" && status != "terminated" {
				allCompleted = false
				break
			}
		}
		
		if allCompleted {
			w.logger.Println("All nodes completed, completing workflow")
			w.completeWorkflow(context)
		}
	}

	// Check if any node failed
	if msg.Status == "failed" {
		w.logger.Printf("Node failed: %s", msg.NodeID)
		w.failWorkflow(context, fmt.Sprintf("Node failed: %s", msg.NodeID))
	}
}

// handleNodeError handles the NodeErrorMessage message
func (w *WorkflowActor) handleNodeError(context actor.Context, msg *NodeErrorMessage) {
	w.mu.Lock()
	defer w.mu.Unlock()

	w.logger.Printf("Node error: %s - %s", msg.NodeID, msg.Error)

	// Update node status
	w.nodeStatuses[msg.NodeID] = "failed"

	// Fail workflow
	w.failWorkflow(context, fmt.Sprintf("Node error: %s - %s", msg.NodeID, msg.Error))
}

// handleDataMessage handles the DataMessage message
func (w *WorkflowActor) handleDataMessage(context actor.Context, msg *system.DataMessage) {
	// Forward message to the appropriate node
	if pid, ok := w.nodeActors[msg.DestinationID]; ok {
		context.Forward(pid)
	} else {
		w.logger.Printf("Unknown destination node: %s", msg.DestinationID)
	}
}

// createNodeActors creates the node actors
func (w *WorkflowActor) createNodeActors(context actor.Context) error {
	w.mu.Lock()
	defer w.mu.Unlock()

	// Create source actors
	for id, source := range w.workflow.Sources {
		props := actor.PropsFromProducer(func() actor.Actor {
			return NewSourceActor(id, source, w.instance.ID, w.stateManager, w.metricsCollector)
		}).WithSupervisor(NewNodeSupervisorStrategy())

		pid, err := context.SpawnNamed(props, fmt.Sprintf("source-%s", id))
		if err != nil {
			return fmt.Errorf("failed to spawn source actor %s: %w", id, err)
		}

		w.nodeActors[id] = pid
		w.nodeStatuses[id] = "created"
		w.logger.Printf("Created source actor: %s", id)
	}

	// Create processor actors
	for id, processor := range w.workflow.Processors {
		props := actor.PropsFromProducer(func() actor.Actor {
			return NewProcessorActor(id, processor, w.instance.ID, w.stateManager, w.metricsCollector)
		}).WithSupervisor(NewNodeSupervisorStrategy())

		pid, err := context.SpawnNamed(props, fmt.Sprintf("processor-%s", id))
		if err != nil {
			return fmt.Errorf("failed to spawn processor actor %s: %w", id, err)
		}

		w.nodeActors[id] = pid
		w.nodeStatuses[id] = "created"
		w.logger.Printf("Created processor actor: %s", id)
	}

	// Create sink actors
	for id, sink := range w.workflow.Sinks {
		props := actor.PropsFromProducer(func() actor.Actor {
			return NewSinkActor(id, sink, w.instance.ID, w.stateManager, w.metricsCollector)
		}).WithSupervisor(NewNodeSupervisorStrategy())

		pid, err := context.SpawnNamed(props, fmt.Sprintf("sink-%s", id))
		if err != nil {
			return fmt.Errorf("failed to spawn sink actor %s: %w", id, err)
		}

		w.nodeActors[id] = pid
		w.nodeStatuses[id] = "created"
		w.logger.Printf("Created sink actor: %s", id)
	}

	return nil
}

// connectNodes connects the nodes according to the workflow edges
func (w *WorkflowActor) connectNodes(context actor.Context) error {
	w.mu.RLock()
	defer w.mu.RUnlock()

	// Connect nodes
	for _, edge := range w.workflow.Edges {
		// Get source node
		var sourcePID *actor.PID
		var sourceID string
		if edge.From.SourceID != "" {
			sourcePID = w.nodeActors[edge.From.SourceID]
			sourceID = edge.From.SourceID
		} else if edge.From.ProcessorID != "" {
			sourcePID = w.nodeActors[edge.From.ProcessorID]
			sourceID = edge.From.ProcessorID
		} else {
			return fmt.Errorf("invalid edge source")
		}

		// Get target node
		var targetPID *actor.PID
		var targetID string
		if edge.To.ProcessorID != "" {
			targetPID = w.nodeActors[edge.To.ProcessorID]
			targetID = edge.To.ProcessorID
		} else if edge.To.SinkID != "" {
			targetPID = w.nodeActors[edge.To.SinkID]
			targetID = edge.To.SinkID
		} else {
			return fmt.Errorf("invalid edge target")
		}

		// Send connect message to source node
		future := context.RequestFuture(sourcePID, &ConnectNodeCommand{
			TargetID: targetID,
			TargetPID: targetPID,
		}, 5*time.Second)

		result, err := future.Result()
		if err != nil {
			return fmt.Errorf("failed to connect nodes: %w", err)
		}

		response, ok := result.(*ConnectNodeResponse)
		if !ok || !response.Success {
			return fmt.Errorf("failed to connect nodes: %s", response.Error)
		}

		w.logger.Printf("Connected nodes: %s -> %s", sourceID, targetID)
	}

	return nil
}

// startSourceNodes starts the source nodes
func (w *WorkflowActor) startSourceNodes(context actor.Context) error {
	w.mu.RLock()
	defer w.mu.RUnlock()

	// Start source nodes
	for id, source := range w.workflow.Sources {
		pid := w.nodeActors[id]

		// Send start message to source node
		future := context.RequestFuture(pid, &StartNodeCommand{
			Parameters: w.instance.Parameters,
		}, 5*time.Second)

		result, err := future.Result()
		if err != nil {
			return fmt.Errorf("failed to start source node: %w", err)
		}

		response, ok := result.(*StartNodeResponse)
		if !ok || !response.Success {
			return fmt.Errorf("failed to start source node: %s", response.Error)
		}

		// Update node status
		w.nodeStatuses[id] = "running"
		w.logger.Printf("Started source node: %s", id)
	}

	return nil
}

// completeWorkflow completes the workflow
func (w *WorkflowActor) completeWorkflow(context actor.Context) {
	// Send completion message to supervisor
	if w.supervisor != nil {
		context.Send(w.supervisor, &WorkflowCompletedMessage{
			RunID: w.instance.ID,
		})
	}

	// Report metrics
	if w.metricsCollector != nil {
		context.Send(w.metricsCollector, &system.ReportMetricCommand{
			Name:  "workflow.completed",
			Value: 1,
			Tags: map[string]string{
				"workflow_id": w.workflow.ID,
				"instance_id": w.instance.ID,
			},
		})
	}

	// Stop self
	context.Self().Stop()
}

// failWorkflow fails the workflow
func (w *WorkflowActor) failWorkflow(context actor.Context, error string) {
	// Send failure message to supervisor
	if w.supervisor != nil {
		context.Send(w.supervisor, &WorkflowFailedMessage{
			RunID: w.instance.ID,
			Error: error,
		})
	}

	// Report metrics
	if w.metricsCollector != nil {
		context.Send(w.metricsCollector, &system.ReportMetricCommand{
			Name:  "workflow.failed",
			Value: 1,
			Tags: map[string]string{
				"workflow_id": w.workflow.ID,
				"instance_id": w.instance.ID,
			},
		})
	}

	// Stop self
	context.Self().Stop()
}

// NewNodeSupervisorStrategy creates a new node supervisor strategy
func NewNodeSupervisorStrategy() actor.SupervisorStrategy {
	return actor.NewOneForOneStrategy(3, 5*time.Second, func(reason interface{}) actor.Directive {
		switch reason.(type) {
		case *ProcessingError:
			return actor.RestartDirective
		case *ResourceExhaustedError:
			return actor.RestartDirective
		default:
			return actor.EscalateDirective
		}
	})
}

// ProcessingError represents a processing error
type ProcessingError struct {
	// Message is the error message
	Message string
	// Cause is the underlying error
	Cause error
}

// Error returns the error message
func (e *ProcessingError) Error() string {
	if e.Cause != nil {
		return e.Message + ": " + e.Cause.Error()
	}
	return e.Message
}

// ResourceExhaustedError represents a resource exhausted error
type ResourceExhaustedError struct {
	// Resource is the exhausted resource
	Resource string
	// Message is the error message
	Message string
}

// Error returns the error message
func (e *ResourceExhaustedError) Error() string {
	return fmt.Sprintf("resource exhausted: %s - %s", e.Resource, e.Message)
}

// NodeStatusChangedMessage is a message for node status changes
type NodeStatusChangedMessage struct {
	// NodeID is the node ID
	NodeID string
	// Status is the new status
	Status string
}

// NodeErrorMessage is a message for node errors
type NodeErrorMessage struct {
	// NodeID is the node ID
	NodeID string
	// Error is the error message
	Error string
}

// ConnectNodeCommand is a command to connect a node
type ConnectNodeCommand struct {
	// TargetID is the target node ID
	TargetID string
	// TargetPID is the target node PID
	TargetPID *actor.PID
}

// ConnectNodeResponse is the response to a ConnectNodeCommand
type ConnectNodeResponse struct {
	// Success indicates if the operation was successful
	Success bool
	// Error is the error message if the operation failed
	Error string
}

// StartNodeCommand is a command to start a node
type StartNodeCommand struct {
	// Parameters are the node parameters
	Parameters map[string]interface{}
}

// StartNodeResponse is the response to a StartNodeCommand
type StartNodeResponse struct {
	// Success indicates if the operation was successful
	Success bool
	// Error is the error message if the operation failed
	Error string
}

// NewSourceActor creates a new source actor
func NewSourceActor(id string, source SourceDefinition, instanceID string, stateManager *actor.PID, metricsCollector *actor.PID) actor.Actor {
	// This is a placeholder implementation
	// The actual implementation will be in a separate file
	return &actor.EmptyActor{}
}

// NewProcessorActor creates a new processor actor
func NewProcessorActor(id string, processor ProcessorDefinition, instanceID string, stateManager *actor.PID, metricsCollector *actor.PID) actor.Actor {
	// This is a placeholder implementation
	// The actual implementation will be in a separate file
	return &actor.EmptyActor{}
}

// NewSinkActor creates a new sink actor
func NewSinkActor(id string, sink SinkDefinition, instanceID string, stateManager *actor.PID, metricsCollector *actor.PID) actor.Actor {
	// This is a placeholder implementation
	// The actual implementation will be in a separate file
	return &actor.EmptyActor{}
}
